/**
 * Xinfei.com Inc.
 * Copyright (c) 2004-2023 All Rights Reserved.
 */
package com.xinfei.techplayprod.biz.mq;

import com.xinfei.xfframework.common.SpringUtil;
import com.xinfei.xfframework.context.Context;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.admin.ConsumeStats;
import org.apache.rocketmq.common.protocol.body.ClusterInfo;
import org.apache.rocketmq.common.protocol.body.ConsumeStatsList;
import org.apache.rocketmq.common.protocol.route.BrokerData;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author Chengsheng.Li
 * @version $ MqLagService, v 0.1 2023/12/1 10:27 Chengsheng.Li Exp $
 */
@Service
@Slf4j
public class MqService {
    private DefaultMQAdminExt mqAdminExt = new DefaultMQAdminExt();

    @PostConstruct
    public void init() {
        setUpMq();
    }

    public String getTopicConsumerGroup() {
        try {
            ClusterInfo clusterInfo = mqAdminExt.examineBrokerClusterInfo();
            Map<String, String> topicConsumerGroup = new HashMap<>();
            Map<String, BrokerData> map = clusterInfo.getBrokerAddrTable();
            for (Map.Entry<String, BrokerData> entry : map.entrySet()) {
                BrokerData brokerData = entry.getValue();
                String brokerAddr = brokerData.getBrokerAddrs().get(MixAll.MASTER_ID);
                ConsumeStatsList consumeStatsList = mqAdminExt.fetchConsumeStatsInBroker(brokerAddr, false, 5000);
                List<Map<String, List<ConsumeStats>>> consumeLst = consumeStatsList.getConsumeStatsList();
                for (Map<String, List<ConsumeStats>> constat : consumeLst) {
                    for (Map.Entry<String, List<ConsumeStats>> stringlstEntry : constat.entrySet()) {
                        String consumer = stringlstEntry.getKey();
                        List<ConsumeStats> consumeStats = stringlstEntry.getValue();
                        for (ConsumeStats consumeStats1 : consumeStats) {
                            consumeStats1.getOffsetTable().entrySet().forEach(t -> {
                                String topic = t.getKey().getTopic();
                                boolean flag = true;
                                if (topic.contains("DLQ") || topic.contains("RETRY") || topic.contains("SYNC_BROKER_MEMBER") || topic.contains("RMQ_SYS") || topic.contains("SCHEDULE_TOPIC") || topic.contains("SELF_TEST_TOPIC") || topic.contains("DefaultCluster") || topic.contains("TBW102") || topic.contains("BenchmarkTest") || topic.contains("broker")) {
                                    flag = false;
                                }
                                if (flag) {
                                    if (!topicConsumerGroup.containsKey(topic)) {
                                        topicConsumerGroup.put(topic, consumer);
                                    } else if(!topicConsumerGroup.get(topic).contains(consumer)){
                                        topicConsumerGroup.put(topic, topicConsumerGroup.get(topic) + "," + consumer);
                                    }
                                }
                            });
                        }
                    }
                }
            }
            StringBuilder sr = new StringBuilder();
            for (Map.Entry<String, String> entry : topicConsumerGroup.entrySet()) {
                sr.append(entry.getKey()).append(":").append(entry.getValue()).append(System.lineSeparator());
            }
            return sr.toString();

        } catch (Throwable e) {
            return "";
        }
    }

    private void setUpMq() {
        String key = "rocketmq.nameServer";
        String nameServer = SpringUtil.getPro(key);
        if (!StringUtils.hasText(nameServer)) {
            if ("local".equalsIgnoreCase(Context.ENV)) {
                nameServer = "dev-mqnameserver.testxinfei.cn:9876";
            } else if (Context.isProd()) {
                nameServer = "mqnameserver.xinfei.io:9876";
            } else {
                nameServer = "dev-mqnameserver.testxinfei.io:9876";
            }
        }
        mqAdminExt.setNamesrvAddr(nameServer);
        try {
            mqAdminExt.start();
        } catch (Throwable e) {
            log.error("", e);
        }
    }
}